package util;

import java.io.File;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TextFileLoader {
	private Queue<String> statusQueue = new ConcurrentLinkedQueue<String>();
	private final int QUEUE_LOWER;
	private boolean eof = false;
	private AtomicBoolean loading = new AtomicBoolean(true);
	private Lock lock = new ReentrantLock();
	private Condition startCond = lock.newCondition();

	private final String filePrefix;
	private int count = -1;

	/**
	 * 那么filePrefix_表示的一个具体的文件
	 * 
	 * @param filePrefix_
	 * @param count_
	 * @param threshold_
	 *            队列中最多允许的元素个数
	 */
	public TextFileLoader(String filePrefix_, int count_, int threshold_) {
		filePrefix = filePrefix_;
		count = count_;
		QUEUE_LOWER = threshold_;
	}

	public void start() {
		new Thread(filePrefix) {
			@Override
			public void run() {
				long fileIndex = 0;
				while (!eof) {
					if (!loop && fileIndex > 0) {
						break;
					} else if (count != -1 && fileIndex >= count) {
						break;
					}

					IFileReader ifr = null;
					if (count != -1) {
						ifr = new IFileReader(new File(filePrefix
								+ (fileIndex % count)));
					} else {
						ifr = new IFileReader(new File(filePrefix));
					}
					fileIndex++;

					ifr.open();
					String line = "";
					while (!eof && (line = ifr.readLine()) != null) {
						statusQueue.offer(line);

						while (statusQueue.size() > QUEUE_LOWER) {
							try {
								lock.lock();
								// leave at most 2000 elements in the queue,
								// then
								// wait 1000 are consumed
								System.err.println("loading suspended");
								loading.set(false);
								startCond.await();
								System.err.println("loading started");
								loading.set(true);
							} catch (InterruptedException e) {
							} finally {
								lock.unlock();
							}
						}

					}
					ifr.close();
				}
				eof = true;
				System.err.println("load thread terminated");
			}
		}.start();

		while (loading.get() || eof) {
			try {
				Thread.sleep(100);
				break;
			} catch (InterruptedException e) {
			}
		}
	}

	boolean loop = true;

	/**
	 * 那么filePrefix_表示的一个具体的文件
	 * 
	 * @param filePrefix_
	 * @param count_
	 */
	public TextFileLoader(String filePrefix_, boolean loop_, int threshold_) {
		filePrefix = filePrefix_;
		loop = loop_;
		QUEUE_LOWER = threshold_;
	}

	public String readLine() {
		String ret = statusQueue.poll();
		do {
			if (ret != null) {
				if (statusQueue.size() < QUEUE_LOWER / 2 && !loading.get()
						&& !eof) {
					try {
						lock.lock();
						startCond.signal();
					} finally {
						lock.unlock();
					}
				}
				break;
			} else {
				if (!loading.get() && !eof) {
					try {
						lock.lock();
						startCond.signal();
					} finally {
						lock.unlock();
					}
				}
			}
			ret = statusQueue.poll();
		} while (!eof);
		return ret;
	}

	public void close() {
		eof = true;
		if (loading.get() == false) {
			lock.lock();
			startCond.signal();
			lock.unlock();
		}
	}

}
